package dse.messages.multiversion;

import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;
import java.util.TreeSet;

import dse.*;
import dse.messages.Message;
import dse.messages.MessagePurpose;
import dse.messages.MessageUtil;
import dse.messages.Transaction;
import dse.messages.singleversion.*;

public class RepErasureTransaction extends Transaction implements
		ShareDownloader, VersionReader, LockProcessor{
	private Node requester;
	private DataObject desiredObject;
	
	private LinkedList<Node> residentNodes;
	private Node lockServer;
	private boolean failed;
	private TreeMap<Version, LinkedList<Node>> sharesDownloaded;
	//private TreeMap<Version, LinkedList<Node>> versionsDownloaded;
	private int requestShareCount;
	private int validShareCount;
	private int maxFailure;
	private int successCount;
	private int failureCount;
	private Version maxVersion;
	private int sharesToBeUpdate;
	private DataRequest request;
	private int maxTS;
	private boolean onoff;
	
	public RepErasureTransaction(Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
		
		this.requester = requester;
		this.lockServer = null;
		this.desiredObject = desiredObject;
		this.failed = false;
		this.requestShareCount = 0;
		this.validShareCount = 0;
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.successCount = 0;
		this.failureCount = 0;
		this.maxVersion = null;
		this.sharesToBeUpdate = desiredObject.shareCount();
		this.request = null;
		this.maxTS = 0;
		this.onoff = false;
		
		//String a = isUpdate?"Update":"Read";
		//System.out.println("Node: " + requester.id() + ", Data: " + desiredObject.id() + ", " + a);
	}

	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		residentNodes = MessageUtil.getNodesByDht(experiment.network(), requester, desiredObject, experiment.modules().assigner());		
		lockServer = experiment.modules().assigner().assign(desiredObject, 1, experiment.network());
		
		if(isUpdate)
		{
			recorder.dataUpdateInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			this.requestShareCount = desiredObject.shareCount();
			this.validShareCount = desiredObject.shareCount()-this.maxFailure;
			//for(Node n : this.residentNodes){
	        //    Message m = super.createRequestVersionRepMessage(this.requester, n, desiredObject, MessagePurpose.DataObjectUpdate, true, this);
	            //recorder.debug(this, "new msg: " + m, time);
	        //    experiment.messageQueue().enqueueMessage(m, startTime);
	        //}
			request = new DataRequest(DataRequest.WRITE, requester.id(), 
					0, this.lockServer, this.id, desiredObject.id());
			request.setProcessor(this);
			sendUpdateLockRequest(recorder, startTime);
		}
		else
		{
			recorder.dataRequestInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			
			this.requestShareCount = this.desiredObject.shareCount();// + this.maxFailure;
			this.validShareCount = desiredObject.requiredShareCount();
			for(int i=0; i<this.requestShareCount; i++)
			{
				Message m = super.createRequestVersionRepMessage(this.requester, this.residentNodes.get(i), desiredObject, MessagePurpose.DataObjectRequest, false, this);
	            //recorder.debug(this, "new msg: " + m, time);
	            experiment.messageQueue().enqueueMessage(m, startTime);
			}	
		}
	}
	
	private void sendUpdateLockRequest(ExperimentRecorder recorder, Double time){
		Message m = super.createRequestUpdateLockMessage(requester, lockServer, desiredObject, request, 
				MessagePurpose.DataObjectUpdate, this);
		//recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	@Override
	public void shareDownloaded (ExperimentRecorder recorder, Node from,
			DataObject dobj, int shareIndex, double timeLeft)
	{
	}
	
	@Override
	public void shareDownloadFailed (ExperimentRecorder recorder, DataObject dobj, 
			int shareIndex, double time)
	{		
	}
	
	@Override
	public boolean isDoneDownloadingShares ()
	{
		return (failed || finished);
	}
	
	/*private void sendGetLatestAndVersionRequest(ExperimentRecorder recorder, Node node, double time, int i)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestLatestAndVersionMessage(requester, node, desiredObject, 
				MessagePurpose.DataObjectRequest, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
		/*if(i == 0)
			begin = m.endTime();
		else if(i == (requestShareCount-1) || i==(this.residentNodes.size()-1)){
			end = m.endTime();
			recorder.readDuration(end - begin);
		}
	}*/
	
	@Override
	public void shareUpdated (ExperimentRecorder recorder, Node to,
			DataObject dobj, int shareIndex, double time)
	{
		if(isUpdate) {		
			this.successCount++;
			if(successCount+failureCount >= this.sharesToBeUpdate)
			{
				onoff = false;
				for(Node n : this.residentNodes){
		            Message m = super.createTurnVersionRepOnMessage(this.requester, n, desiredObject, MessagePurpose.DataObjectUpdate, this.maxTS, onoff, this);
		            //recorder.debug(this, "new msg: " + m, time);
		            experiment.messageQueue().enqueueMessage(m, time);
		        }
				if(this.successCount >= validShareCount)
					failed = false;
				else
					failed = true;
				this.successCount = 0;
				this.failureCount = 0;				
			}
		}
	}
	
	@Override
	public void shareUpdateFailed (ExperimentRecorder recorder, DataObject dobj, 
			int shareIndex, double time)
	{
		if(isUpdate) {
			this.failureCount++;
			if(successCount+failureCount >= this.sharesToBeUpdate)
			{
				onoff = false;
				for(Node n : this.residentNodes){
		            Message m = super.createTurnVersionRepOnMessage(this.requester, n, desiredObject, MessagePurpose.DataObjectUpdate, this.maxTS, onoff, this);
		            //recorder.debug(this, "new msg: " + m, time);
		            experiment.messageQueue().enqueueMessage(m, time);
		        }
				if(this.successCount >= validShareCount)
					failed = false;
				else
					failed = true;
				this.successCount = 0;
				this.failureCount = 0;
			}
		}
	}
	
	@Override
	public boolean isDoneUpdatingShares ()
	{
		return (failed || finished);
	}
	
	private void sendUpdateShareRequest(ExperimentRecorder recorder, Node node, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createShareUpdateV3Message(requester, node, desiredObject, 
				maxVersion, MessagePurpose.DataObjectUpdate, this);
		//recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	@Override
	public void readVersion(ExperimentRecorder recorder, TreeSet<Version> versions, double time, Node from){
		/*if(failed)
			return;
		if(versions == null)
		{
			failureCount++;
			if(failureCount>maxFailure && !failed)
			{
				System.out.println("Failed number exceeds the maximum number");
				transactionFailed(recorder, time);
			}
			return;
		}
		//System.out.println("version number list : " + versions.size());
		recorder.shareHistorySize(versions.size());
		if(versionsDownloaded == null)
			versionsDownloaded = new TreeMap<Version, LinkedList<Node>>();
		//successCount++;
		for(Version ver : versions)
		{
			if(this.versionsDownloaded.get(ver) == null){
				this.versionsDownloaded.put(ver, new LinkedList<Node>());
				this.versionsDownloaded.get(ver).add(from);
			}
			else
				this.versionsDownloaded.get(ver).add(from);
		}*/
	}
	public void readVersion(ExperimentRecorder recorder, int seq, int index, double time){
		this.successCount++;
		if(successCount >= requestShareCount){
			successCount = 0;
			failureCount = 0;
			if(onoff){
				this.maxVersion = new Version(this.maxTS, requester.id());
				for(Node n : this.residentNodes){
		            this.sendUpdateShareRequest(recorder, n, time);
		        }
			}
			else{
				if(!finished){
					Message m = super.createReleaseUpdateLockMessage(requester, lockServer, desiredObject,
						MessagePurpose.DataObjectUpdate);
					//recorder.debug(this, "new msg: " + m, time);
					experiment.messageQueue().enqueueMessage(m, time);
					if(!failed){
						this.transactionSucceed(recorder, time);
					}
					else{
						this.transactionFailed(recorder, time);
					}
				}
			}
		}

	}
	public void readVersion(ExperimentRecorder recorder, Node from, Version version, double time)
	{
		if(failed)
			return;
		if(version == null){
			failureCount++;
			return;
		}
			
		if(sharesDownloaded == null)
			sharesDownloaded = new TreeMap<Version, LinkedList<Node>>();
		successCount++;
		if(this.sharesDownloaded.get(version) == null)
		{
			this.sharesDownloaded.put(version, new LinkedList<Node>());
			this.sharesDownloaded.get(version).add(from);
		}
		else
			this.sharesDownloaded.get(version).add(from);
		
		if(successCount + failureCount >= requestShareCount)
			processVersion(recorder, time);
	}
	
	public void readVersion(ExperimentRecorder recorder, int seq, double time)
	{
		if(seq == -1)
		{
			failureCount++;
			return;
		}
		successCount++;
		if(this.maxTS < seq)
			this.maxTS = seq;
		if(successCount+failureCount >= requestShareCount){
			successCount = 0;
			failureCount = 0;
			
			if(isUpdate){
				this.maxTS++;
				onoff = true;
				for(Node n : this.residentNodes){
		            Message m = super.createTurnVersionRepOnMessage(this.requester, n, desiredObject, MessagePurpose.DataObjectUpdate, this.maxTS, onoff, this);
		            //recorder.debug(this, "new msg: " + m, time);
		            experiment.messageQueue().enqueueMessage(m, time);
		        }
			}
			else{
				for(int i=0; i<this.requestShareCount; i++)
				{
					Message m = super.createRequestDesignatedShareRepMessage(this.requester, this.residentNodes.get(i), desiredObject, this.maxTS, MessagePurpose.DataObjectRequest, this);
		            //recorder.debug(this, "new msg: " + m, time);
		            experiment.messageQueue().enqueueMessage(m, time);
				}	
			}
		}
	}
	public void readVersion(ExperimentRecorder recorder, double time, LinkedList<Version> versions, Version version, Node from){}
		
	/*private void processVersions(ExperimentRecorder recorder, double timeLeft)
	{
		if((versionsDownloaded==null) || (versionsDownloaded.size()==0))
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		Map.Entry<Version, LinkedList<Node>> version = versionsDownloaded.pollLastEntry();
		while((versionsDownloaded.size()>0) && (version.getValue().size() < desiredObject.requiredShareCount()))
		{
			version = versionsDownloaded.pollLastEntry();
		}
		maxVersion = version.getKey();
		LinkedList<Node> availNodes = version.getValue();
		this.requestShareCount = desiredObject.requiredShareCount();
//		this.successCount = 0;
//		this.failureCount = 0;
		if(availNodes.size() < this.requestShareCount){
			System.err.println("No data available");
			this.transactionFailed(recorder, timeLeft);
		}
		else{
			for(int i=0; i<this.requestShareCount; i++)
				SendRequestDesignatedShareRequest(recorder, maxVersion, availNodes.get(i), timeLeft);
		}
	}
	
	private void SendRequestDesignatedShareRequest(ExperimentRecorder recorder, Version version, Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestDesignatedShareMessage(requester, node, desiredObject, version,
				MessagePurpose.DataObjectRequest, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}*/
	
	private void processVersion(ExperimentRecorder recorder, double timeLeft)
	{
		if((sharesDownloaded==null) || (sharesDownloaded.size()==0))
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		
		maxVersion = sharesDownloaded.lastKey();
		if(maxVersion.getSequence()>requester.getSequence())
			requester.setSequence(maxVersion.getSequence());
		Map.Entry<Version, LinkedList<Node>> vn = sharesDownloaded.pollLastEntry();
		while((vn.getValue().size()<validShareCount) && (sharesDownloaded.size()>0))
		{
			vn = sharesDownloaded.pollLastEntry();
		}
		
		if(vn.getValue().size() == requestShareCount)
		{
			transactionSucceed(recorder, timeLeft);
		}
		else if(vn.getValue().size() < requestShareCount && vn.getValue().size()>=validShareCount)
		{
			maxVersion = vn.getKey();
			for(Version version : sharesDownloaded.keySet())
			{
				for(Node node : sharesDownloaded.get(version))
				{
					sendUpdateShareRequest(recorder, node, timeLeft);
				}
			}
			transactionSucceed(recorder, timeLeft);	
		}
		/*else if(sharesDownloaded.size() == 0)
		{	
			//int srf = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject)
				//+ this.maxFailure*2;
			failureCount = 0;
			successCount = 0;
			
			//this.round++;
			this.sharesDownloaded.clear();
			processVersions(recorder, timeLeft);
			//for(int i=0; (i<srf)&&(i<residentNodes.size()); i++)
			//if(this.id == 178)
				//System.out.println("get previous than " + maxVersion.toString());
			//for(int i=0; (i<requestShareCount)&&(i<residentNodes.size()); i++)	
				//sendGetPreviousShareRequest(recorder, maxVersion, residentNodes.get(i), timeLeft);
		}*/
		else
			transactionFailed(recorder, timeLeft);
	}
	
	private void transactionSucceed(ExperimentRecorder recorder, double time)
	{
		finished = true;
		failed  = false;
		//System.out.println("Data ID : " + desiredObject.id() + "Version : " + maxVersion.getSequence() + " , " + maxVersion.getClientID());
		if(isUpdate)
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(super.id, requester, desiredObject, true, time);
		}
		else
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataRequestFinished(super.id, requester, desiredObject, true, this.residentNodes.get(this.validShareCount-1),
					false, time);
		}
	}
	
	private void transactionFailed (ExperimentRecorder recorder, double time)
	{
		recorder.debug(this, "transFailed", time);
		
		finished = true;
		failed = true;
		if(isUpdate)
			recorder.dataUpdateFinished(super.id, requester, desiredObject, false, time);
		else
			recorder.dataRequestFinished(super.id, requester, desiredObject, false, null,
				false, time);
	}
	
	public void lockProcess(boolean success, Node from, ExperimentRecorder recorder, double time){
		for(Node n : this.residentNodes){
            Message m = super.createRequestVersionRepMessage(this.requester, n, desiredObject, MessagePurpose.DataObjectUpdate, true, this);
            //recorder.debug(this, "new msg: " + m, time);
            experiment.messageQueue().enqueueMessage(m, time);
        }
	}
	
	public LinkedList<Node> getFollowers(){
		return null;
	}
	public Node getRequester(){
		return this.requester;
	}
	public Transaction getTransaction(){
		return this;
	}
}
